package logic

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/zeromicro/go-zero/core/logx"
	"mrshen_rocket_comsumer/common/constant"
	"mrshen_rocket_comsumer/internal/config"
	"mrshen_rocket_comsumer/internal/svc"
)

type HelloRocketConsumer struct {
	c      config.Config
	cxt    context.Context
	svcCtx *svc.ServiceContext
}

func NewHelloRocketConsumer(svcCtx *svc.ServiceContext, cxt context.Context, c config.Config) *HelloRocketConsumer {
	return &HelloRocketConsumer{
		c:      c,
		cxt:    cxt,
		svcCtx: svcCtx,
	}
}

func (l *HelloRocketConsumer) HelloRocketConsumer() error {
	err := l.svcCtx.PushHelloGroupConsumer.Subscribe(constant.TASK_ROKET_TOPIC, consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for i := range msgs {
			fmt.Printf("subscribe callback1: %v \n", string(msgs[i].Body))
			// 查询
			list, err := l.svcCtx.TRegionOpenCityModel.FindAll(l.cxt)

			if err != nil {
				logx.Error("查询数据失败")
			}

			for _, item := range list {
				data, _ := json.Marshal(&item)
				logx.Info("用户信息为:  ", string(data))
			}

			logx.Info("接收到消息1: ", string(msgs[i].Body))
		}
		return consumer.ConsumeSuccess, nil
	})

	err = l.svcCtx.PushHelloGroupConsumer1.Subscribe(constant.TASK_ROKET_TOPIC, consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for i := range msgs {
			fmt.Printf("subscribe callback2: %v \n", string(msgs[i].Body))
			logx.Info("接收到消息2: ", string(msgs[i].Body))
		}
		return consumer.ConsumeSuccess, nil
	})

	err = l.svcCtx.PushHelloGroupConsumer2.Subscribe(constant.TASK_ROKET_TOPIC, consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for i := range msgs {
			fmt.Printf("subscribe callback3: %v \n", string(msgs[i].Body))
			logx.Info("接收到消息3: ", string(msgs[i].Body))
		}
		return consumer.ConsumeSuccess, nil
	})

	if err != nil {
		logx.Error(err)
	}
	// 开启消费
	err = l.svcCtx.PushHelloGroupConsumer.Start()
	err = l.svcCtx.PushHelloGroupConsumer1.Start()
	err = l.svcCtx.PushHelloGroupConsumer2.Start()
	if err != nil {
		logx.Error(err)
	}
	return nil
}
